package cn.com.bluemoon.bd.flink.runner;

import cn.com.bluemoon.bd.flink.creater.env.StreamEnvCreater;
import cn.com.bluemoon.bd.flink.enumtype.StateBackendType;
import cn.com.bluemoon.bd.flink.pojo.EnvConfig;
import cn.com.bluemoon.bd.flink.udf.UdfRegister;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.io.File;
import java.time.Duration;

import static cn.com.bluemoon.bd.flink.common.CheckpointConstants.*;
import static cn.com.bluemoon.bd.flink.common.Constants.*;

public class FlinkRunner {
    private AbstractProcessorBuilder builder;
    private ParameterTool parameterTool;
    private EnvConfig envConfig;
    private Duration stateTtl;
    TimeCharacteristic timeCharacteristic;

    public FlinkRunner(AbstractProcessorBuilder builder) {
        this.builder = builder;
        parameterTool = ParameterTool.fromArgs(builder.args);
        stateTtl = Duration.ofHours(parameterTool.getInt("state_ttl_hour", 25));
        long checkpointMs = parameterTool.getInt(CHECKPOINT_TIME, 1800000);
        String checkpointMode = parameterTool.get(CHECKPOINT_MODE, CHECKPOINT_MODE_EXACTLY_ONCE);
        CheckpointingMode checkpointingMode = checkpointMode.equals(CHECKPOINT_MODE_AT_LEAST_ONCE) ? CheckpointingMode.AT_LEAST_ONCE : CheckpointingMode.EXACTLY_ONCE;
        long checkpointInterval = parameterTool.getLong(CHECKPOINT_INTERVAL, checkpointMs / 2);
        long checkpointTimeout = parameterTool.getLong(CHECKPOINT_TIMEOUT, 1800000);
        int checkpointFailureNum = parameterTool.getInt(CHECKPOINT_FAILURE_NUM, 5);
        String checkpointPath = this.createCheckpointPath();
        int checkpointCleanUp = parameterTool.getInt(CHECKPOINT_CLEANUP, 0);
        String timeCharac = parameterTool.get(TIME_CHARACTERISTIC, PROCESSING_TIME);
        timeCharacteristic = timeCharac.equalsIgnoreCase(EVENT_TIME) ? TimeCharacteristic.EventTime : TimeCharacteristic.ProcessingTime;
        StateBackendType stateBackendType = parameterTool.get(STATE_BACKEND_TYPE, "rocksdb").equals("hashmap") ? StateBackendType.HASHMAP : StateBackendType.ROCKSDB;
        int changelogEnable = parameterTool.getInt(CHANGELOG_ENABLE, 0);
        envConfig = new EnvConfig(checkpointingMode, checkpointPath, checkpointMs, checkpointInterval, checkpointTimeout, checkpointFailureNum, checkpointCleanUp, stateBackendType, changelogEnable);
    }

    public void run() throws Exception {
        //初始化环境变量
        StreamExecutionEnvironment env = StreamEnvCreater.create(envConfig);
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        TableConfig tableEnvConfig = tableEnv.getConfig();
        tableEnvConfig.setIdleStateRetention(stateTtl);
        //设置时间特征
        env.setStreamTimeCharacteristic(timeCharacteristic);
        //注册udf函数
        UdfRegister.register(tableEnv);
        //逻辑处理构造器
        builder.build(tableEnv, env).process();
        //执行
        int envExecute = parameterTool.getInt(ENV_EXECUTE, 1);

        if (envExecute != 0) {
            env.execute();
        }
    }


    /**
     * 生成checkpoint hdfs全路径
     *
     * @return
     * @throws Exception
     */
    public String createCheckpointPath() {
        String userDir = System.getProperty("user.dir");

        if (userDir.endsWith(File.separator + "bin")) {
            userDir = userDir.substring(0, userDir.lastIndexOf(File.separator + "bin"));
        }
        String projectName = userDir.substring(userDir.lastIndexOf(File.separator) + 1);
        String jobName = parameterTool.get("job_name", projectName);
        return HDFS_STATEBACKEND_ROOT_PATH + "/jobs/" + jobName + "/checkpoints";
    }
}
